package org.jheaps.tree

import org.jheaps.AddressableHeap
import org.jheaps.MergeableAddressableHeap
import org.jheaps.annotations.ConstantTime
import org.jheaps.annotations.LogarithmicTime
import java.io.Serializable
import kotlin.math.max

/**
 * Rank-Pairing heaps. The heap is sorted according to the
 * [natural ordering][Comparable] of its keys, or by a
 * [Comparator] provided at heap creation time, depending on which
 * constructor is used.
 *
 *
 *
 * This is the type-1 rank-pairing heap described in detail in the following
 * [paper](https://doi.org/10.1137/100785351):
 *
 *  * B Haeupler, S Sen, RE Tarjan. Rank-Pairing Heaps. SIAM Journal of
 * Computing, 40(6): 1463--1485, 2011.
 *
 *
 *
 *
 * This implementation provides amortized O(1) time for operations that do not
 * involve deleting an element such as `insert`, and `decreaseKey`.
 * Operations `deleteMin` and `delete` are amortized O(log(n)). The
 * operation `meld` is also amortized O(1).
 *
 *
 *
 * All the above bounds, however, assume that the user does not perform
 * cascading melds on heaps such as:
 *
 * <pre>
 * d.meld(e);
 * c.meld(d);
 * b.meld(c);
 * a.meld(b);
</pre> *
 *
 * The above scenario, although efficiently supported by using union-find with
 * path compression, invalidates the claimed bounds.
 *
 *
 *
 * Note that the ordering maintained by this heap, like any heap, and whether or
 * not an explicit comparator is provided, must be *consistent with
 * `equals`* if this heap is to correctly implement the
 * `AddressableHeap` interface. (See `Comparable` or
 * `Comparator` for a precise definition of *consistent with
 * equals*.) This is so because the `AddressableHeap` interface is
 * defined in terms of the `equals` operation, but this heap performs all
 * key comparisons using its `compareTo` (or `compare`) method, so
 * two keys that are deemed equal by this method are, from the standpoint of
 * this heap, equal. The behavior of a heap *is* well-defined even if its
 * ordering is inconsistent with `equals`; it just fails to obey the
 * general contract of the `AddressableHeap` interface.
 *
 *
 *
 * **Note that this implementation is not synchronized.** If
 * multiple threads access a heap concurrently, and at least one of the threads
 * modifies the heap structurally, it *must* be synchronized externally.
 * (A structural modification is any operation that adds or deletes one or more
 * elements or changing the key of some element.) This is typically accomplished
 * by synchronizing on some object that naturally encapsulates the heap.
 *
 * @param <K>
 * the type of keys maintained by this heap
 * @param <V>
 * the type of values maintained by this heap
 *
 * @author Dimitrios Michail
 *
 * @see RankPairingHeap
 *
 * @see CostlessMeldPairingHeap
 *
 * @see FibonacciHeap
</V></K> */
class RankPairingHeap<K, V> @ConstantTime constructor(
    /**
     * The comparator used to maintain order in this heap, or null if it uses
     * the natural ordering of its keys.
     *
     * @serial
     */
    private val comparator: Comparator<in K?>?
) : MergeableAddressableHeap<K,V>,
    Serializable {

    /**
     * The last node in the root list
     */
    private var minRoot: Node? = null

    /**
     * Size of the pairing heap
     */
    private var size: Long

    /**
     * Auxiliary array for consolidation.
     */
    private val aux: Array<Node?>

    /**
     * Used to reference the current heap or some other pairing heap in case of
     * melding, so that handles remain valid even after a meld, without having
     * to iterate over them.
     *
     * In order to avoid maintaining a full-fledged union-find data structure,
     * we disallow a heap to be used in melding more than once. We use however,
     * path-compression in case of cascading melds, that it, a handle moves from
     * one heap to another and then another.
     */
    private var other: RankPairingHeap<K, V>

    /**
     * Constructs a new, empty heap, using the natural ordering of its keys. All
     * keys inserted into the heap must implement the [Comparable]
     * interface. Furthermore, all such keys must be *mutually
     * comparable*: `k1.compareTo(k2)` must not throw a
     * `ClassCastException` for any keys `k1` and `k2` in the
     * heap. If the user attempts to put a key into the heap that violates this
     * constraint (for example, the user attempts to put a string key into a
     * heap whose keys are integers), the `insert(Object key)` call will
     * throw a `ClassCastException`.
     */
    @ConstantTime
    constructor() : this(null)

    /**
     * Constructs a new, empty heap, ordered according to the given comparator.
     * All keys inserted into the heap must be *mutually comparable* by
     * the given comparator: `comparator.compare(k1,
     * k2)` must not throw a `ClassCastException` for any keys `k1`
     * and `k2` in the heap. If the user attempts to put a key into the
     * heap that violates this constraint, the `insert(Object key)` call
     * will throw a `ClassCastException`.
     *
     * @param comparator
     * the comparator that will be used to order this heap. If
     * `null`, the [natural ordering][Comparable] of
     * the keys will be used.
     */
    init {
        this.size = 0
        this.aux = java.lang.reflect.Array.newInstance(Node::class.java, AUX_BUCKET_ARRAY_SIZE) as Array<Node?>
        this.other = this
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @ConstantTime(amortized = true)
    override fun insert(key: K, value: V?): Node {
        check(other === this) { "A heap cannot be used after a meld" }
        if (key == null) {
            throw NullPointerException("Null keys not permitted")
        }
        val n = Node(this, key, value)
        if (minRoot == null) {
            n.r = n
            minRoot = n
        } else {
            n.r = minRoot!!.r
            minRoot!!.r = n
            if (less(n, minRoot)) {
                minRoot = n
            }
        }
        size++
        return n
    }

    /**
     * {@inheritDoc}
     *
     * @throws IllegalStateException
     * if the heap has already been used in the right hand side of a
     * meld
     */
    @ConstantTime(amortized = true)
    override fun insert(key: K):Node{
        return insert(key, null)
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun findMin(): Node {
        if (size == 0L) {
            throw NoSuchElementException()
        }
        return minRoot!!
    }

    /**
     * {@inheritDoc}
     */
    @LogarithmicTime(amortized = true)
    override fun deleteMin(): Node {
        if (size == 0L) {
            throw NoSuchElementException()
        }

        /*
         * Sever spine of left child of minimum
         */
        val oldMinRoot = minRoot
        var spine: Node? = null
        if (minRoot!!.l != null) {
            spine = severSpine(minRoot!!.l)
            minRoot!!.l = null
        }

        /*
         * One pass spine
         */
        var maxRank = -1
        var output: Node? = null
        while (spine != null) {
            // remove second from list
            var cur: Node
            if (spine.r === spine) {
                cur = spine
                spine = null
            } else {
                cur = spine.r!!
                spine.r = cur.r
            }
            cur.r = null

            val rank = cur.rank
            val auxEntry = aux[rank]
            if (auxEntry == null) {
                aux[rank] = cur
                if (rank > maxRank) {
                    maxRank = rank
                }
            } else {
                aux[rank] = null
                cur = link(cur, auxEntry)
                // add to output list
                if (output == null) {
                    cur.r = cur
                    output = cur
                } else {
                    cur.r = output.r
                    output.r = cur
                    if (less(cur, output)) {
                        // keep track of new minimum
                        output = cur
                    }
                }
            }
        }

        /*
         * One pass old half-trees, careful to skip old minimum which is still
         * in the root list.
         */
        while (minRoot != null) {
            // remove second from list
            var cur: Node
            if (minRoot!!.r === minRoot) {
                cur = minRoot!!
                minRoot = null
            } else {
                cur = minRoot!!.r!!
                minRoot!!.r = cur.r
            }
            cur.r = null

            if (cur === oldMinRoot) {
                // skip the old minimum
                continue
            }

            val rank = cur.rank
            val auxEntry = aux[rank]
            if (auxEntry == null) {
                aux[rank] = cur
                if (rank > maxRank) {
                    maxRank = rank
                }
            } else {
                aux[rank] = null
                cur = link(cur, auxEntry)
                // add to output list
                if (output == null) {
                    cur.r = cur
                    output = cur
                } else {
                    cur.r = output.r
                    output.r = cur
                    if (less(cur, output)) {
                        // keep track of new minimum
                        output = cur
                    }
                }
            }
        }

        /*
         * Process remaining in buckets
         */
        for (i in 0..maxRank) {
            val cur = aux[i]
            if (cur != null) {
                aux[i] = null
                // add to output list
                if (output == null) {
                    cur.r = cur
                    output = cur
                } else {
                    cur.r = output.r
                    output.r = cur
                    if (less(cur, output)) {
                        // keep track of new minimum
                        output = cur
                    }
                }
            }
        }

        minRoot = output
        size--
        return oldMinRoot!!
    }

    @get:ConstantTime
    override val isEmpty: Boolean
        /**
         * {@inheritDoc}
         */
        get() = size == 0L

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun size(): Long {
        return size
    }

    /**
     * {@inheritDoc}
     */
    override fun comparator(): Comparator<in K?>? {
        return comparator
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime(amortized = false)
    override fun clear() {
        minRoot = null
        size = 0
    }

    /**
     * {@inheritDoc}
     */
    @ConstantTime
    override fun meld(other: MergeableAddressableHeap<K,V>) {
        val h = other as RankPairingHeap<K,V>

        // check same comparator
        if (comparator != null) {
            require(!(h.comparator == null || h.comparator != comparator)) { "Cannot meld heaps using different comparators!" }
        } else require(h.comparator == null) { "Cannot meld heaps using different comparators!" }

        check(h.other === h) { "A heap cannot be used after a meld." }

        // meld
        if (minRoot == null) {
            minRoot = h.minRoot
        } else if (h.minRoot != null) {
            val afterMinRoot = minRoot!!.r
            val hAfterMinRoot = h.minRoot!!.r

            minRoot!!.r = hAfterMinRoot
            h.minRoot!!.r = afterMinRoot
            if (less(h.minRoot, minRoot)) {
                minRoot = h.minRoot
            }
        }
        size += h.size

        // clear other
        h.size = 0
        h.minRoot = null

        // take ownership
        h.other = this
    }

    // --------------------------------------------------------------------
    inner class Node(/*
         * We maintain explicitly the belonging heap, instead of using an inner
         * class due to possible cascading melding.
         */
          var heap: RankPairingHeap<K, V>,
          override var key: K,
          override var value: V?
    ) : AddressableHeap.Handle<K, V>, Serializable {
        /**
         * {@inheritDoc}
         */
        var p: Node? = null // parent
        var l: Node? = null // left child
        var r: Node? = null // right child or next root
        var rank: Int = 0

        /**
         * {@inheritDoc}
         */
        @ConstantTime(amortized = true)
        override fun decreaseKey(newKey: K) {
            owner.decreaseKey(this, newKey)
        }

        /**
         * {@inheritDoc}
         */
        @LogarithmicTime(amortized = true)
        override fun delete() {
            require(!(p == null && r == null)) { "Invalid handle!" }
            val h = owner
            h.forceDecreaseKeyToMinimum(this)
            h.deleteMin()
        }

        /*
       * Get the owner heap of the handle. This is union-find with
       * path-compression between heaps.
       */
        val owner: RankPairingHeap<K, V>
            get() {
                if (heap.other !== heap) {
                    // find root
                    var root = heap
                    while (root !== root.other) {
                        root = root.other
                    }
                    // path-compression
                    var cur = heap
                    while (cur.other !== root) {
                        val next = cur.other
                        cur.other = root
                        cur = next
                    }
                    heap = root
                }
                return heap
            }
    }

    /*
     * Decrease the key of a node to the minimum. Helper function for performing
     * a delete operation. Does not change the node's actual key, but behaves as
     * the key is the minimum key in the heap.
     */
    private fun forceDecreaseKeyToMinimum(n: Node) {
        // already at root list
        if (n.p == null) {
            minRoot = n
            return
        }

        // perform the cut
        val u = n.p
        cut(n)
        if (minRoot == null) {
            n.r = n
        } else {
            n.r = minRoot!!.r
            minRoot!!.r = n
        }
        minRoot = n

        // restore type-1 ranks
        n.rank = if ((n.l == null)) 0 else n.l!!.rank + 1
        restoreType1Ranks(u)
    }

    /**
     * Decrease the key of a node.
     *
     * @param n
     * the node
     * @param newKey
     * the new key
     */
    private fun decreaseKey(n: Node, newKey: K) {
        val c = comparator?.compare(newKey, n.key) ?: (newKey as Comparable<K?>).compareTo(n.key)
        require(c <= 0) { "Keys can only be decreased!" }

        n.key = newKey
        if (c == 0) {
            return
        }

        require(!(n.p == null && n.r == null)) { "Invalid handle!" }

        // already at root list, just update minimum if needed
        if (n.p == null) {
            if (less(n, minRoot)) {
                minRoot = n
            }
            return
        }

        // perform the cut
        val u = n.p
        cut(n)
        if (minRoot == null) {
            n.r = n
            minRoot = n
        } else {
            n.r = minRoot!!.r
            minRoot!!.r = n
            if (less(n, minRoot)) {
                minRoot = n
            }
        }

        n.rank = if ((n.l == null)) 0 else n.l!!.rank + 1
        restoreType1Ranks(u)
    }

    /*
     * Return the minimum node
     */
    private fun less(x: Node?, y: Node?): Boolean {
        return if (comparator == null) {
            (x!!.key as Comparable<K?>?)!!.compareTo(y!!.key) < 0
        } else {
            comparator.compare(x!!.key, y!!.key) < 0
        }
    }

    /*
     * Iterate over all right children of x and create a list of half-tree
     * roots. We only need to nullify the parent pointers. At the same time give
     * each new root a rank that is one greater than that of its left child.
     */
    private fun severSpine(x: Node?): Node? {
        var cur = x
        while (cur!!.r != null) {
            cur.p = null
            if (cur.l == null) {
                cur.rank = 0
            } else {
                cur.rank = cur.l!!.rank + 1
            }
            cur = cur.r
        }
        cur.p = null
        if (cur.l == null) {
            cur.rank = 0
        } else {
            cur.rank = cur.l!!.rank + 1
        }
        cur.r = x
        return x
    }

    /*
     * Link two half-trees. Assumes that x and y are the roots.
     */
    private fun link(x: Node, y: Node): Node {
        assert(x.rank == y.rank)
        val c = comparator?.compare(x.key, y.key) ?: (x.key as Comparable<K?>?)!!.compareTo(y.key)

        if (c <= 0) {
            y.r = x.l
            if (x.l != null) {
                x.l!!.p = y
            }
            x.l = y
            y.p = x
            x.rank += 1
            return x
        } else {
            x.r = y.l
            if (y.l != null) {
                y.l!!.p = x
            }
            y.l = x
            x.p = y
            y.rank += 1
            return y
        }
    }

    /*
     * Cut x and its left child from its parent. Leave the right child of x in
     * place of x.
     */
    private fun cut(x: Node) {
        val u = x.p!!
        val y = x.r
        if (u.l === x) {
            u.l = y
        } else {
            u.r = y
        }
        if (y != null) {
            y.p = u
        }
        x.p = null
        x.r = x
    }

    /*
     * Restore the type-1 ranks after a cut.
     * 
     * @param The old parent of the node which was cut.
     */
    private fun restoreType1Ranks(u: Node?) {
        var u = u
        while (u != null) {
            val leftRank = if ((u.l == null)) -1 else u.l!!.rank
            if (u.p == null) {
                u.rank = leftRank + 1
                break
            }
            val rightRank = if ((u.r == null)) -1 else u.r!!.rank
            val k = if ((leftRank == rightRank)) leftRank + 1 else max(leftRank.toDouble(), rightRank.toDouble())
                .toInt()
            if (k >= u.rank) {
                break
            }
            u.rank = k
            u = u.p
        }
    }

    companion object {
        /**
         * Size of bucket array. Based on maximum rank.
         */
        private const val AUX_BUCKET_ARRAY_SIZE = 65
    }
}
